%%%-------------------------------------------------------------------
%%% File    : rdf_store.erl
%%% Author  :  <vjache>
%%% Description : 
%%%
%%% Created : 26 May 2009 by  <vjache>
%%%-------------------------------------------------------------------
-module(rdf_store).

%% Includes
-include("../include/util.hrl").
-include("../include/iter.hrl").
-include("../include/pipe.hrl").
-include("../include/rdf_record.hrl").
-include("../include/rdf_spalg.hrl").
-include("../include/rrs.hrl").

%% API
-export([
		 open/1,
		 add/2,
		 remove/2,
		 find/3,
		 find_direct/3,
		 find_directc/3,
		 count/1,
		 count/2,
		 clear/1,
		 upload/2,
		 index_for/1
		]).
-export([
		 test/0,
		 test/1
		]).

-record(session,{name}).

%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
open(Name) ->
	%Check exists
    #session{name=Name}.
%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
add(_Sess,QuadSource) when is_record(QuadSource, iter) ->
	mapred:exec(
	  {map,
	   fun(?SPOG=Spog) ->
			   [{S,Spog},
				{P,?PSOG},
				{O,?OSPG},
				{G,?GSPO}];
		  (BadRec) ->
			   exit({bad_quad,BadRec,"Iterator must contain tuples in form {spog,S,P,O,G}."})
	   end,
	   progress_support(QuadSource),
	   [{zip,1000}]
	  }, pipe_vect:new(vect_by_tag_list()));
add(Sess,QuadSource) when is_list(QuadSource)->
    add(Sess,iter:from_list(QuadSource)).
vect_by_tag_list() ->
	[{_Tag=rdf_record:tag_by_role(Role),
	  _Vect=index_for(Role)}||Role<-?ALL_ROLEs].
%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
upload(Sess,{log,FilePath}) when is_list(FilePath)->
	add(Sess,iter_disk_log:new([{name,make_ref()},{file,FilePath}]));
upload(Sess,{tup,FilePath}) when is_list(FilePath)->
	Iter=iter_mapper:new(iter_term_reader:new_from_file(FilePath), 
					fun({?SPOG_TAG,_S,_P,_O,_G}=Rec) ->
							Rec;
					   ({S,P,O}) ->
							{?SPOG_TAG,S,P,O,_G={FilePath}};
					   (BadRec) ->
							   exit({bad_spog,BadRec,at,FilePath,"File must contain tuples in form {spog,S,P,O,G} or {S,P,O}."})
					end),
	add(Sess,Iter).

%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
remove(Sess,QuadSource) when is_record(QuadSource, iter)->
    ok;
remove(Sess,QuadSource) when is_list(QuadSource)->
    ok.
%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
find(Sess,QuadPattSpec,Destination) 
  when is_record(Sess,session)
  andalso is_record(Destination, pipe) ->
	case lookup_info(QuadPatt=normalize(QuadPattSpec)) of
		undefined ->
			exit({too_wide_quad_pattern,QuadPatt});
		{Node,Role,_Const} ->
			case node() of
				Node ->
					ok=find_directc(Role,QuadPatt,Destination);
				_ ->
					ok=rpc:call(Node, ?MODULE, find_directc, [Role,QuadPatt,Destination])
			end
	end;
find(Sess,QuadPattSpec,{local,Pipe}=_Destination) 
  when is_record(Sess,session)
  andalso is_record(Pipe, pipe)->
	case lookup_info(QuadPatt=normalize(QuadPattSpec)) of
		undefined ->
			exit({too_wide_quad_pattern,QuadPatt});
		{Node,Role,_Const} ->
			case node() of
				Node ->
					find_direct(Role,QuadPatt,Pipe);
				_ ->
					PipeRes=pipe_result:new(ResultId=make_ref(), self()),
					MonitorPid=spawn_link(
								 fun()->
										 ok=rpc:call(Node, ?MODULE, find_directc, [Role,QuadPatt,PipeRes])
								 end),
					pipe_result:fold(
					  MonitorPid, ResultId, 
					  fun(Rec,PipeState) ->
							  pipe:push(PipeState,Rec)
					  end, pipe:open(Pipe))
			end
	end.

find_directc(Role,QuadPatt,Pipe) ->
	PipeState=find_direct(Role,QuadPatt,Pipe),
	ok=pipe:close(PipeState).
find_direct(Role,QuadPatt,Pipe) ->
	QuadPattNorm=normalize(QuadPatt),
	case element(rdf_record:role_pos(Role)+1,QuadPattNorm) of
		Const when ?IS_CONST(Const) ->
			BestIndex=rdf_sol_builder:index_for(Role),
			VectSess=vect:open(BestIndex),
			Iter=iter_mapper:new(
				   vect:lookup(VectSess,Const), 
				   fun(Quad) ->
						   rdf_sol_builder:match(Quad, QuadPattNorm)
				   end),
			{IterState,PipeState}=pipe:push_each(Pipe, Iter),
			ok=iter:close(IterState),
			PipeState;
		BadConst ->
			exit({not_a_const,BadConst,at_role,Role,at_patt,QuadPattNorm})
	end.

lookup_info(QuadPatt) when is_record(QuadPatt, ?QUAD_TAG_SA) ->
	case best_lookup(QuadPatt) of
		undefined ->
			undefined;
		{Role,Const} ->
			Node=?ROUTE2UPNODE(svc_rrs:key_route(Const)),
			{Node,Role,Const}
	end.
%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
clear(Sess) ->
	ok.
%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
count(Sess,QuadPatt) when is_record(QuadPatt, ?QUAD_TAG_SA) ->
    ok.
%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
count(Sess) ->
    ok.
%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
spalg_select(Sess,Destination) when 
  is_record(Destination, pipe)->
    ok;
spalg_select(Sess,{local,Pipe}=_Destination) when 
  is_record(Pipe, pipe)->
    ok;
spalg_select(Sess,{Node,Pipe}=_Destination) when 
  is_record(Pipe, pipe) ->
    ok.
%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
spalg_constr(Sess,Destination) when 
  is_record(Destination, pipe)->
    ok;
spalg_constr(Sess,{local,Pipe}=_Destination) when 
  is_record(Pipe, pipe)->
    ok;
spalg_constr(Sess,{Node,Pipe}=_Destination) when 
  is_record(Pipe, pipe) ->
    ok.

%%====================================================================
%% Internal functions
%%====================================================================
best_lookup({?QUAD_TAG_SA, S,_P,_O,_G}) when ?IS_CONST(S)->
	{s,S};
best_lookup({?QUAD_TAG_SA, _S,_P,O,_G}) when ?IS_CONST(O)->
	{o,O};
best_lookup({?QUAD_TAG_SA, _S,_P,_O,G}) when ?IS_CONST(G)->
	{g,G};
best_lookup({?QUAD_TAG_SA, _S,P,_O,_G}) when ?IS_CONST(P)->
	{p,P};
best_lookup({?QUAD_TAG_SA, _S,_P,_O,_G}) ->
	undefined.


normalize(QuadPattProto) when is_record(QuadPattProto, ?QUAD_TAG_SA) ->
	QuadPattProto;
normalize(QuadPattProto) when tuple_size(QuadPattProto)<5 ->
	case QuadPattProto of
		{S,P,O,G} ->
			{?QUAD_TAG_SA,S,P,O,G};
		{S,P,O} ->
			{?QUAD_TAG_SA,S,P,O,'_'};
		{S,P} ->
			{?QUAD_TAG_SA,S,P,'_','_'};
		{S} ->
			{?QUAD_TAG_SA,S,'_','_','_'}
	end;
normalize(QuadPattProto) when is_list(QuadPattProto) ->
	InitList=[{1,?QUAD_TAG_SA}|[{rdf_record:role_pos(Role)+1,Val}||{Role,Val}<-QuadPattProto]],
	erlang:make_tuple(5, '_', InitList).

progress_support(Iter) when is_record(Iter, iter)->
	Prefix="RDF-STORE ADD: ",
	Postfix=" AT [node=~p, pid=~p, time=~p]~n",
	StartMsg=Prefix++"START"++Postfix,
	ProgressMsg=Prefix++"DONE ~p"++Postfix,
	FinishMsg=Prefix++"FINISHED WITH ~p"++Postfix,
	iter_counter:new_progress_observer(
	  Iter, 1000,
	  fun(start,_Count)->
			  io:format(StartMsg,[node(),self(),now()]);
		 (progress,Count)->
			  io:format(ProgressMsg,[Count,node(),self(),now()]);
		 (finish,Count)->
			  io:format(FinishMsg,[Count,node(),self(),now()])
	  end).

index_for(ValueRole) ->
	case svc_vms:get_vect(rdf_record:tag_by_role(ValueRole)) of
		{error,_}=ERR ->
			exit(ERR);
		Vect ->
			Vect
	end.

%%====================================================================
%% Test functions
%%====================================================================

test(find) ->
	Sess=rdf_store:open("TestStore"),rdf_store:find(Sess, [{s,{onto,comp_0}}], {local,pipe_console:new()}),
%% 	find(Sess, [{g,{"d:/_Development/Erlang/wsp/demetra/test/rdf_store_SUITE_data/companies.tup"}}], {local,pipe_console:new()}),
	
	ok.
test() ->
	Sess=open("TestStore"),
	upload(Sess,{log,"d:/_Development/Erlang/wsp/demetra/bin/mega_data_demetra1@ws24292.LOG"}).

